package com.csw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo14Actions {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("actions")
      .setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    /**
      *
      * 转换算子：从一个RDD转换成另一个RDD，是RDD之前的转换，
      * 转换算子是懒执行，需要action算子触发执行
      *
      * 操作算子：触发任务执行，每一个action算子都会触发一个job任务
      *
      */

    /**
      *
      * 构建RDD的方式
      * 1、读取文件
      * 2、基于scala集合构建RDD
      *
      */

    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    /**
      * count：统计rdd的行数
      * 会触发任务执行
      */
    println(studentRDD.count())

    /**
      * foreach：遍历数据，一般只能用于本地测试
      * foreachPartition：一次处理一个分区的数据， 一般用于将数据保存到外部的时候
      */
    //    studentRDD.foreach(println)
    //
    //    studentRDD.foreachPartition(i=>{
    //      println("一个分区")
    //      i.foreach(println)
    //    })

    /**
      * collect：将rdd转换成集合，将rdd的数据拉取到内存中，如果rdd数据量比较大会导致内存溢出
      */
    val array: Array[String] = studentRDD.collect()

    array.foreach(println)

    /**
      * reduce：全局聚合
      *
      * 类似于SQL:select sum(age) fromstudents;
      */
    val ageRDD: RDD[Int] = studentRDD.map(i => i.split(",")(2).toInt)

    val sumAge: Int = ageRDD.reduce((x, y) => x + y)

    println(sumAge)

    /**
      * save：将数据保存到hdfs中，如果输出目录已经存在会报错
      */
    studentRDD.saveAsTextFile("/spark/data/out1")

  }
}
